package com.dtsheep.rxjava.stduy.observable;

import rx.Observable;
import rx.Subscriber;
import rx.Subscription;
import rx.schedulers.Schedulers;

import java.util.Objects;

/**
 * FilterDemo
 *
 * <p>
 *     过滤元素
 * </p>
 *
 * @author chuanbai
 * @date 2024/5/17
 */
public class FilterDemo {

    /**
     * filter 过滤元素
     *
     * @param
     * @return
     * @author chuanbai
     * @date ${DATE}
     */
    public void  filter01(){
        // 创建 被观察对象 、 发射器 、 调度器
        Observable<String> observable = Observable.unsafeCreate((Subscriber<? super String> emmiter) -> {
            emmiter.onNext("11111");
            emmiter.onNext("test");
            emmiter.onNext("小猫叫");
            emmiter.onCompleted();
        }).subscribeOn(Schedulers.computation());

        // map 处理元素
        Observable<String> obsMap = observable.map(obj -> obj);

        // filter 过滤出值为 “11111” 的元素
        Observable<String> obsFilter = obsMap.filter(obj -> Objects.equals(obj, "11111"));

        // 观察者订阅
        Subscription subscription = obsFilter.subscribe(obj -> {
            System.out.println(obj);
        });

        // 取消订阅
        subscription.unsubscribe();

        // 查看是否取消订阅
        boolean unsubscribed = subscription.isUnsubscribed();
        System.out.println(unsubscribed);

    }

    public static void main(String[] args) {
        FilterDemo filterDemo = new FilterDemo();
        filterDemo.filter01();
    }
}
